{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyalink.alink import *\n",
    "useLocalEnv(4)\n",
    "\n",
    "from utils import *\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "DATA_DIR = ROOT_DIR + \"mnist\" + os.sep\n",
    "\n",
    "DENSE_TRAIN_FILE = \"dense_train.ak\";\n",
    "SPARSE_TRAIN_FILE = \"sparse_train.ak\";\n",
    "\n",
    "INIT_MODEL_FILE = \"init_model.ak\";\n",
    "TEMP_STREAM_FILE = \"temp_stream.ak\";\n",
    "\n",
    "VECTOR_COL_NAME = \"vec\";\n",
    "LABEL_COL_NAME = \"label\";\n",
    "PREDICTION_COL_NAME = \"cluster_id\";\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1\n",
    "dense_source = AkSourceBatchOp().setFilePath(DATA_DIR + DENSE_TRAIN_FILE);\n",
    "sparse_source = AkSourceBatchOp().setFilePath(DATA_DIR + SPARSE_TRAIN_FILE);\n",
    "sw = Stopwatch();\n",
    "\n",
    "pipelineList = [\n",
    "    [\"KMeans EUCLIDEAN\",\n",
    "     Pipeline()\\\n",
    "        .add(\n",
    "            KMeans()\\\n",
    "                .setK(10)\\\n",
    "                .setVectorCol(VECTOR_COL_NAME)\\\n",
    "                .setPredictionCol(PREDICTION_COL_NAME)\n",
    "        )\n",
    "    ],\n",
    "    [\"KMeans COSINE\",\n",
    "     Pipeline()\\\n",
    "        .add(\n",
    "            KMeans()\\\n",
    "                .setDistanceType('COSINE')\\\n",
    "                .setK(10)\\\n",
    "                .setVectorCol(VECTOR_COL_NAME)\\\n",
    "                .setPredictionCol(PREDICTION_COL_NAME)\n",
    "        )\n",
    "    ],\n",
    "    [\"BisectingKMeans\",\n",
    "     Pipeline()\\\n",
    "        .add(\n",
    "            BisectingKMeans()\\\n",
    "                .setK(10)\\\n",
    "                .setVectorCol(VECTOR_COL_NAME)\\\n",
    "                .setPredictionCol(PREDICTION_COL_NAME)\n",
    "        )\n",
    "    ]\n",
    "]\n",
    "\n",
    "for pipelineTuple2 in pipelineList :\n",
    "    sw.reset();\n",
    "    sw.start();\n",
    "    pipelineTuple2[1]\\\n",
    "    .fit(dense_source)\\\n",
    "    .transform(dense_source)\\\n",
    "    .link(\n",
    "        EvalClusterBatchOp()\\\n",
    "            .setVectorCol(VECTOR_COL_NAME)\\\n",
    "            .setPredictionCol(PREDICTION_COL_NAME)\\\n",
    "            .setLabelCol(LABEL_COL_NAME)\\\n",
    "            .lazyPrintMetrics(pipelineTuple2[0] + \" DENSE\")\n",
    "    );\n",
    "    BatchOperator.execute();\n",
    "    sw.stop();\n",
    "    print(sw.getElapsedTimeSpan());\n",
    "\n",
    "    sw.reset();\n",
    "    sw.start();\n",
    "    pipelineTuple2[1]\\\n",
    "        .fit(sparse_source)\\\n",
    "        .transform(sparse_source)\\\n",
    "        .link(\n",
    "            EvalClusterBatchOp()\\\n",
    "                .setVectorCol(VECTOR_COL_NAME)\\\n",
    "                .setPredictionCol(PREDICTION_COL_NAME)\\\n",
    "                .setLabelCol(LABEL_COL_NAME)\\\n",
    "                .lazyPrintMetrics(pipelineTuple2[0] + \" SPARSE\")\n",
    "        );\n",
    "    BatchOperator.execute();\n",
    "    sw.stop();\n",
    "    print(sw.getElapsedTimeSpan());\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2\n",
    "batch_source = AkSourceBatchOp().setFilePath(DATA_DIR + SPARSE_TRAIN_FILE);\n",
    "stream_source = AkSourceStreamOp().setFilePath(DATA_DIR + SPARSE_TRAIN_FILE);\n",
    "\n",
    "if not(os.path.exists(DATA_DIR + INIT_MODEL_FILE)) :\n",
    "    batch_source\\\n",
    "        .sampleWithSize(100)\\\n",
    "        .link(\n",
    "            KMeansTrainBatchOp()\\\n",
    "                .setVectorCol(VECTOR_COL_NAME)\\\n",
    "                .setK(10)\n",
    "        )\\\n",
    "        .link(\n",
    "            AkSinkBatchOp()\\\n",
    "                .setFilePath(DATA_DIR + INIT_MODEL_FILE)\n",
    "        );\n",
    "    BatchOperator.execute();\n",
    "\n",
    "\n",
    "init_model = AkSourceBatchOp().setFilePath(DATA_DIR + INIT_MODEL_FILE);\n",
    "\n",
    "KMeansPredictBatchOp()\\\n",
    "    .setPredictionCol(PREDICTION_COL_NAME)\\\n",
    "    .linkFrom(init_model, batch_source)\\\n",
    "    .link(\n",
    "        EvalClusterBatchOp()\\\n",
    "            .setVectorCol(VECTOR_COL_NAME)\\\n",
    "            .setPredictionCol(PREDICTION_COL_NAME)\\\n",
    "            .setLabelCol(LABEL_COL_NAME)\\\n",
    "            .lazyPrintMetrics(\"Batch Prediction\")\n",
    "    );\n",
    "BatchOperator.execute();\n",
    "\n",
    "stream_source\\\n",
    "    .link(\n",
    "        KMeansPredictStreamOp(init_model)\\\n",
    "            .setPredictionCol(PREDICTION_COL_NAME)\n",
    "    )\\\n",
    "    .link(\n",
    "        AkSinkStreamOp()\\\n",
    "            .setFilePath(DATA_DIR + TEMP_STREAM_FILE)\\\n",
    "            .setOverwriteSink(True)\n",
    "    );\n",
    "StreamOperator.execute();\n",
    "\n",
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + TEMP_STREAM_FILE)\\\n",
    "    .link(\n",
    "        EvalClusterBatchOp()\\\n",
    "            .setVectorCol(VECTOR_COL_NAME)\\\n",
    "            .setPredictionCol(PREDICTION_COL_NAME)\\\n",
    "            .setLabelCol(LABEL_COL_NAME)\\\n",
    "            .lazyPrintMetrics(\"Stream Prediction\")\n",
    "    );\n",
    "BatchOperator.execute();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_3\n",
    "\n",
    "pd.set_option('display.html.use_mathjax', False)\n",
    "\n",
    "stream_source = AkSourceStreamOp().setFilePath(DATA_DIR + SPARSE_TRAIN_FILE);\n",
    "\n",
    "init_model = AkSourceBatchOp().setFilePath(DATA_DIR + INIT_MODEL_FILE);\n",
    "\n",
    "stream_pred = stream_source\\\n",
    "    .link(\n",
    "        StreamingKMeansStreamOp(init_model)\\\n",
    "            .setTimeInterval(1)\\\n",
    "            .setHalfLife(1)\\\n",
    "            .setPredictionCol(PREDICTION_COL_NAME)\n",
    "    )\\\n",
    "    .select(PREDICTION_COL_NAME + \", \" + LABEL_COL_NAME +\", \" + VECTOR_COL_NAME);\n",
    "\n",
    "stream_pred.sample(0.001).print();\n",
    "\n",
    "stream_pred\\\n",
    "    .link(\n",
    "        AkSinkStreamOp()\\\n",
    "            .setFilePath(DATA_DIR + TEMP_STREAM_FILE)\\\n",
    "            .setOverwriteSink(True)\n",
    "    );\n",
    "StreamOperator.execute();\n",
    "\n",
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + TEMP_STREAM_FILE)\\\n",
    "    .link(\n",
    "        EvalClusterBatchOp()\\\n",
    "            .setVectorCol(VECTOR_COL_NAME)\\\n",
    "            .setPredictionCol(PREDICTION_COL_NAME)\\\n",
    "            .setLabelCol(LABEL_COL_NAME)\\\n",
    "            .lazyPrintMetrics(\"StreamingKMeans\")\n",
    "    );\n",
    "BatchOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
